package server

import (
	"../../deps/lessgo/utilx"
	"../base"
	"../conf"
	"../net"
	"../state"
	"../store"
	"strconv"
	"strings"
	"time"
)

const (
	learnerRangeSize uint64 = 100
)

type Learner int

type LearnRequest struct {
	Bucket       string
	VersionStart uint64
	VersionStop  uint64
	VersionList  string
}

type LearnReply struct {
	Data []base.DataNode
}

func LearnerTrackerStart() {

	go func() {

		for {

			verGlobal, _ := store.Cmd("GET", "sys:keeper:ver").Uint64()
			time.Sleep(6e9)

			if verGlobal < 1 || len(state.Peers) < 2 {
				continue
			}

			bs, _ := store.NodeGet("/")

			for _, v := range bs.N {

				if v.T != store.NodeTypeDir {
					continue
				}

				bucket := strings.Trim(v.P, "/")

				verLearn, _ := store.Cmd("GET", "is:"+bucket+":lrn:ver").Uint64()

				verLast, verStart, verStop := verLearn, verLearn, verLearn+learnerRangeSize

				for {

					mergeVers := []string{}

					//
					rsls, _ := store.Cmd("ZRANGEBYSCORE", "lg:"+bucket,
						verStart, verStop, "WITHSCORES").List()
					if len(rsls)%2 == 0 {

						for i := 0; i < len(rsls); i += 2 {
							mergeVers = append(mergeVers, rsls[i+1])
							ver, _ := strconv.ParseUint(rsls[i+1], 10, 64)
							if uint64(ver) > verLast {
								verLast = uint64(ver)
							}
						}
					}

					for _, p := range state.Peers {

						if p.Id == conf.Config.HostId {
							continue
						}

						req := LearnRequest{
							Bucket:       bucket,
							VersionStart: verStart,
							VersionStop:  verStop,
							VersionList:  strings.Join(mergeVers, ","),
						}

						call := net.NewNetCall()
						call.Method = "Learner.LogMerge"
						call.Addr = p.Addr + ":" + p.Port
						call.Args = req
						call.Reply = new(LearnReply)

						net.RpcInstance.Call(call)
						<-call.Status

						rpl := call.Reply.(*LearnReply)
						for _, n := range rpl.Data {

							locVer, _ := store.Cmd("ZSCORE", "lg:"+bucket, n.Key).Uint64()
							if n.Version <= locVer {
								continue
							}

							if n.Value == store.NodeDelMagic {
								store.Cmd("SADD", "lg:"+bucket, n.Version, n.Key)
							} else {

								nSet := store.NodeProposal{
									Key:    "/" + bucket + "/" + n.Key,
									Val:    n.Value,
									VerSet: n.Version,
								}
								store.NodeSet(&nSet)
							}

							mergeVers = append(mergeVers, strconv.FormatUint(n.Version, 10))
							if n.Version > verLast {
								verLast = n.Version
							}
						}
					}

					if verLast > verLearn {
						store.Cmd("SET", "is:"+bucket+":lrn:ver", verLast)
					}

					if verStop > verGlobal {
						break
					}

					verStart, verStop = verStop, verStop+learnerRangeSize
				}
			}

		}
	}()
}

func (l *Learner) LogMerge(args *LearnRequest, rep *LearnReply) error {

	vers := strings.Split(args.VersionList, ",")

	ls, _ := store.Cmd("ZRANGEBYSCORE", "lg:"+args.Bucket,
		args.VersionStart, args.VersionStop, "WITHSCORES").List()

	rep.Data = []base.DataNode{}

	if len(ls)%2 == 0 {

		for i := 0; i < len(ls); i += 2 {

			if utilx.ArrayContain(ls[i+1], vers) {
				continue
			}

			node := base.DataNode{
				Key: ls[i], // NOTICE! no bucket name at this
			}

			key := args.Bucket + "/" + ls[i]
			rs, err := store.Cmd("HMGET", store.NodeData+key,
				store.NodeSepRev, store.NodeSepVal).List()
			if err == nil {
				ver, _ := strconv.ParseUint(rs[0], 10, 64)
				node.Value = rs[1]
				node.Version = uint64(ver)
			} else {
				// version == 0 means that the key has been deleted
				node.Version = 0
			}

			if node.Version == 0 || node.Value == store.NodeDirMagic {
				ver, _ := strconv.ParseUint(rs[i+1], 10, 64)
				node.Version = uint64(ver)
				node.Value = store.NodeDelMagic
			}

			rep.Data = append(rep.Data, node)
		}
	}

	return nil
}
